package com.bkjk.mongoToMysql;

import java.io.File;
import java.io.IOException;
import java.io.UnsupportedEncodingException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Optional;
import java.util.Queue;
import java.util.Set;

import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.plugins.PluginRegistry;
import org.pentaho.di.core.plugins.StepPluginType;
import org.pentaho.di.trans.TransHopMeta;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.StepMeta;
import org.pentaho.di.trans.steps.mongodbinput.MongoDbInputMeta;
import org.pentaho.di.trans.steps.tableoutput.TableOutputMeta;
import org.pentaho.mongo.wrapper.field.MongoField;

import com.bkjk.constant.KettleConstant;
import com.bkjk.constant.MongoConstant;
import com.bkjk.dbConn.MongoDao;
import com.bkjk.utils.KettleUtils;

/**
 * 生成非自增表的ktr
 * 
 * 转换问题：mysql中数据是大小写不敏感的 解决方案:将mysql设置为大小写敏感的 建库时排序规则选择utf_bin
 * 
 * system.indexes表未完成 sessions_表未完成
 * 
 * @author liangjie
 *
 */
public class GeneNonIncreKtr{

	private MongoDao mongoDao = MongoDao.getInstance();;

	public GeneNonIncreKtr() {
		// 初次执行，需要初始化kettle相关的env
		try {
			KettleUtils.initEnv();
			// 保证目录的存在
			File file = new File(KettleConstant.NONINCRE_FILE_DIR);
			if (!file.exists()) {
				file.mkdir();
			}
		} catch (KettleException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	/**
	 * 针对mongo中的每一个非自增的表遍历
	 */
	public void transForEveryCollection() {
		mongoDao.init_NonOrIncreCollection("countly");
		// 获取对象
		Queue<String> collections = mongoDao.getNon_increaseCollections();
		// 记录转换出错的表
		Map<String, String> err = new HashMap<>();
		// 针对每一个队列
		TransMeta transMeta = null;
		for (String collectionName : collections) {
			// 处理掉system.indexes中得'.',交在读取中执行
			transMeta = createTransMeta(collectionName);
			Optional<TransMeta> result = Optional.ofNullable(transMeta);
			result.ifPresent(value -> {
				try {
					writeToXML(value, collectionName);
					KettleUtils.executeTrans(value);
				} catch (Exception e) {
					// TODO Auto-generated catch block
					// e.printStackTrace();
					err.put(collectionName, e.getMessage());
				}
			});
		}

		//
		//return err;
	}
	/**
	 * 将ktr文件以流方式输出到xml
	 * @param transMeta
	 * @param collectionName
	 * @throws UnsupportedEncodingException
	 * @throws IOException
	 * @throws KettleException
	 */
	public void writeToXML(TransMeta transMeta, String collectionName)
			throws UnsupportedEncodingException, IOException, KettleException {
		String fileName = KettleConstant.NONINCRE_FILE_DIR + collectionName + "转换" + KettleConstant.FILE_SUFFIX;
		KettleUtils.writeToXML(transMeta, fileName);
	}

	/**
	 * 生成kettle的krt文件
	 * @param collectionName
	 * @return
	 */
	public TransMeta createTransMeta(String collectionName) {
		// 针对每一个创建一个转换
		// Trans trans = KettleUtils.getTrans();
		// 针对每一个转换，创建一个转换meta
		TransMeta transMeta = KettleUtils.getTransMeta();

		// 获取集合中的所有列
		Set<String> columns = mongoDao.getColumnsInCollection(collectionName, "countly");
		// 如果队列为空，那么创建一个空的输出
		if (columns.size() == 0) {
			return null;
		}
		// 设置trans的transMeta
		// trans.setTransMeta(transMeta);

		// 针对每一个转换meta，创建一个Mongo输入
		MongoDbInputMeta mongoDbInputMeta = CustomPlugin.getCustomMongoInput();
		// 设置集合的名称
		mongoDbInputMeta.setCollection(collectionName);

		// 用来设置到mysql的表输出字段
		String[] fieldDatabases = new String[columns.size()];
		// 用来设置到mysql的表输出字段对应的mongo输入流字段
		String[] fieldStream = new String[columns.size()];

		List<MongoField> mongoFields = new ArrayList<>();
		MongoField mongField = null;
		int index = 0;
		for (String column : columns) {
			mongField = new MongoField();
			mongField.m_fieldName = column;
			mongField.m_fieldPath = column;
			mongField.m_kettleType = "String";
			// mongField.setIndexedVals(column);
			mongoFields.add(mongField);
			// 将对应的列设置到fieldDatabases和fieldStream中
			fieldDatabases[index] = KettleUtils.mapColumnsToNonKey(column);
			fieldStream[index] = column;
			index++;
		}
		// 设置mongo不以单独的json输出方式输出
		mongoDbInputMeta.setOutputJson(false);

		mongoDbInputMeta.setMongoFields(mongoFields);
		mongoDbInputMeta.setSocketTimeout(MongoConstant.SOCKET_TIMEOUT);
		// 设置mongo的输出
		// mongoDbInputMeta.setMongoFields();
		// List<MongoField> mongoFields = mongoDbInputMeta.getMongoFields();
		// mongoDbInputMeta.
		// mongoFields.stream().forEach(field->System.out.println(field.""));
		// 创建特定的stepMeta，已经设置了mongoInput的stepID
		StepMeta from_Mongo_stepMeta = CustomPlugin.getCustomMongoStepMeta();
		from_Mongo_stepMeta.setName("mongo集合[" + collectionName + "]");
		from_Mongo_stepMeta.setStepMetaInterface(mongoDbInputMeta);
		// 将该步加入到转换中
		transMeta.addStep(from_Mongo_stepMeta);

		//
		PluginRegistry registry = PluginRegistry.getInstance();
		// 创建一个表输出
		DatabaseMeta dbm = KettleUtils.getDataBase();
		dbm.setName("mysql测试");
		// 将db加入到trans中
		transMeta.addDatabase(dbm);
		// dbm.setObjectId(id);

		TableOutputMeta tableOutput = new TableOutputMeta();
		// 设置是否裁剪表
		tableOutput.setTruncateTable(true);
		tableOutput.setDatabaseMeta(dbm);
		// 主要是处理system.indexes
		String collectionN = collectionName.replaceAll("\\.", "");
		tableOutput.setTableName(collectionN);
		// 设置指定数据库字段
		tableOutput.setSpecifyFields(true);
		// 获取mongo的collection的列，并将输入的列映射到其它的列

		tableOutput.setFieldDatabase(fieldDatabases);
		tableOutput.setFieldStream(fieldStream);
		String tostepid = registry.getPluginId(StepPluginType.class, tableOutput);
		// 新建一个步骤
		StepMeta to_stepMeta = KettleUtils.getStepMeta();
		to_stepMeta.setLocation(400, 100);
		to_stepMeta.setStepMetaInterface(tableOutput);
		to_stepMeta.setName("["+collectionName+"]"+"到mysql[" + collectionN + "]的输出");
		to_stepMeta.setStepID(tostepid);
		// 将步骤添加
		transMeta.addStep(to_stepMeta);
		// 建立一个连接

		TransHopMeta hi = new TransHopMeta(from_Mongo_stepMeta, to_stepMeta);
		transMeta.addTransHop(hi);

		// 创建一个mysql的表输出

		return transMeta;

	}

	public static void doNonIncre() {

	}

	public static void main(String[] args) {
		System.out.println("测试开始");
		try {
			KettleUtils.initEnv();
			String collectionName = "cities";
			String fileName = KettleConstant.FILE_DIR + collectionName + "转换" + KettleConstant.FILE_SUFFIX;
			GeneNonIncreKtr test = new GeneNonIncreKtr();
			TransMeta transMeta = test.createTransMeta(collectionName);

			KettleUtils.writeToXML(transMeta, fileName);
			System.out.println("文件路径为:" + fileName);
		} catch (Exception e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}

		System.out.println("测试结束");
	}

	public void close() {
		if (mongoDao != null) {
			mongoDao.close();
		}
	}
}
